package com.qyer.search.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.TupleUtils;
import com.qyer.search.storm.logging.ErrorLogger;
import com.qyer.search.storm.logging.InfoLogger;
import com.qyer.search.storm.logging.TimeCostLogger;
import com.qyer.search.storm.logging.WarnLogger;
import com.qyer.search.storm.util.ConfigUtils;
import com.qyer.search.storm.util.EntityTags;
import com.qyer.search.storm.util.Tag;
import com.qyer.search.storm.util.ZTag;
import com.qyer.search.storm.util.ZTag2;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tangzheng on 16/7/6.
 */
public class FetchZTagBolt extends JDBCBolt{

  private static final long serialVersionUID = -1L;
  private String _streamId;

  private PreparedStatement _pstmt3;
  private PreparedStatement _pstmt4;
  private final String _other_Sql_Pre;
  private final String _other_Sql_Pre2;

  public FetchZTagBolt(String tagBoltPrefix, int inCount, int inCount2, String streamId,
                       Map<String, String> sqlConfig, String other, String other2){
    super(tagBoltPrefix, inCount, inCount2, sqlConfig);
    this._streamId = streamId;
    this._other_Sql_Pre = other;
    this._other_Sql_Pre2 = other2;
  }

  @Override public void prepare(Map map, TopologyContext topologyContext,
                                OutputCollector outputCollector){

    String directory = "/Data/search/ZTagBolt/";
    String filename = directory + topologyContext.getStormId()+"/"
      + topologyContext.getThisComponentId() +"/"+topologyContext.getThisTaskId();
    String loggername = JDBCBolt.class.getSimpleName() + "@" + topologyContext.getThisTaskId();
    InfoLogger tmpLog = new InfoLogger(filename + "_tmp", loggername + "_tmp");
    dataLog = new InfoLogger(filename + "_data", loggername + "_data");
    infoLogger = new InfoLogger(filename + "_info", loggername + "_info");
    errorLogger = new ErrorLogger(filename +"_error", loggername + "_error");
    timeCostLogger = new TimeCostLogger(filename + "_timecost", loggername + "_timecost");
    successLogger = new WarnLogger(filename + "_success", loggername + "_success");
    failLogger = new WarnLogger(filename + "_fail", loggername + "_fail");
    dbInfoLogger = new InfoLogger(filename + "_db", loggername + "_db");
    tickInfoLogger = new InfoLogger(filename + "_tick", loggername + "_tick");
    processLogger = new InfoLogger(filename + "_process", filename + "_process");


    _topologyContext = topologyContext;
    _outputCollector = outputCollector;
    _config = (Map<String, Object>)map;
    _taskId = topologyContext.getThisTaskId();
    _componentId = topologyContext.getThisComponentId();
    _prefix = _boltPrefix + ",TaskId:"+ _taskId + ",ComponentId:" + _componentId;

    try {
      _conn = ConfigUtils.getConnection((Map<String, Object>) _sqlConfig, _boltPrefix);
      dbInfoLogger.info("{} DB init successfully!", _boltPrefix);
    }catch (Exception e){
      dbInfoLogger.info("{} Connect DB failed.", _boltPrefix);
    }
    /*
    **select a.lid,a.tid,b.tag_name from (select lid,tid from qyer_lastminute.zworld_lastminute_tag
    * where lid in (?,?,?,?,?,?,?,?,?,?) and isdel = 0) a join qyer_lastminute.zworld_tag as b on a.tid = b.id;
    *
    * //2/
    * select c.lid,d.id,d.tag_name from (select a.lid,b.parent_id from (select lid,tid from
    * qyer_lastminute.zworld_lastminute_tag where lid in (?,?,?,?,?,?,?,?,?,?) and isdel = 0) a
    * join (select from qyer_lastminute.zworld_stype where type = 'stype' and status = 1) b on a.tid = b.id) c
    * join qyer_lastminute.zworld_tag as d on c.parent_id = d.id;
     */
    try {
      _pstmt = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn);
      _pstmt2 = ConfigUtils.getPrepareStmt2((Map<String, Object>) map, _boltPrefix, this._conn);
      _pstmt3 = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn,
                                           this._other_Sql_Pre);
      _pstmt4 = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn,
                                           this._other_Sql_Pre2);
    }catch (Exception e){
      dbInfoLogger.info("{} get PreparedStatement failed.", _boltPrefix);
    }
  }

  @Override
  public void checkDBConnection(){
    try {
      if((_conn == null) || (!_conn.isValid(2))){
        dbInfoLogger.info("mysqlconnection is invalid,rebuild follow.");
        Connection _tempConn = _conn;
        this._conn = ConfigUtils.getConnection(_sqlConfig, _boltPrefix);
        this._pstmt = ConfigUtils.getPrepareStmt(_config, _boltPrefix, _conn);
        this._pstmt2 = ConfigUtils.getPrepareStmt2(_config, _boltPrefix, _conn);
        this._pstmt3 = ConfigUtils.getPrepareStmt(_config, _boltPrefix, _conn, _other_Sql_Pre);
        this._pstmt4 = ConfigUtils.getPrepareStmt(_config, _boltPrefix, _conn, _other_Sql_Pre2);
        if(_tempConn != null) {
          _tempConn.close();
          dbInfoLogger.info("close the past mysqlconnection");
        }
      }else {
        infoLogger.info("mysqlconnection is ok");
      }
    }catch (Exception e){
      e.printStackTrace();
      dbInfoLogger.info("mysqlconnection rebuild failed by {}.", e);
      dbInfoLogger.info("Exception:{},DB connection invalid.", e.getMessage());
    }
  }

  @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(_streamId, new Fields("index", "type", "ztags", "msgId", "flag"));
  }

  @Override public void execute(Tuple tuple) {
    if (TupleUtils.isTick(tuple)) {
      tickInfoLogger.info("Tick arrive.");
      checkDBConnection();
    }else {
      checkDBConnection();
      long now = System.currentTimeMillis();
      String index = tuple.getString(0);
      String type = tuple.getString(1);
      List<Integer> payload = (List<Integer>) tuple.getValue(2);
      int msgCount = tuple.getInteger(3);
      int stage = tuple.getInteger(4);
      PreparedStatement pstmt_in_use;
      PreparedStatement pstmt_in_use_other;
      int inCount_in_use = 0;
      //********************************************************************
      dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.IN, msgCount, index, type, stage, payload);
      //********************************************************************

      pstmt_in_use = (ConfigUtils.INIT_STAGE == stage) ? _pstmt : _pstmt2;
      pstmt_in_use_other = (ConfigUtils.INIT_STAGE == stage) ? _pstmt3 : _pstmt4;
      inCount_in_use = (ConfigUtils.INIT_STAGE == stage) ? _inCount : _inCount2;
      try {
        ConfigUtils.constructSQL(pstmt_in_use, payload, inCount_in_use);
        ConfigUtils.constructSQL(pstmt_in_use_other, payload, inCount_in_use);
      } catch (Exception e) {
        dbInfoLogger.info("test..msgId:{},constructSQL failed,{}", msgCount, e);
        e.printStackTrace();
      }

      int flag = 1;

      int lid;
      int tid;
      int channel = -1;
      String tag_name;

      Map<Integer, List<ZTag2>> tagsMap = new HashMap<>();

      ResultSet rs = null;
      ResultSet rs1 = null;
      try {
        rs = pstmt_in_use.executeQuery();
        while (rs.next()) {
          lid = rs.getInt(1);
          tid = rs.getInt(2);
          tag_name = rs.getString(3);

          ZTag2 zTag = new ZTag2(tid, tag_name, -1);
          processLogger
            .info("msgId:{},product id:{},tid:{},tag_name:{}", msgCount, lid, tid, tag_name);

          if (tagsMap.containsKey(lid)) {
            tagsMap.get(lid).add(zTag);
          } else {
            List<ZTag2> lt = new ArrayList<>();
            lt.add(zTag);
            tagsMap.put(lid, lt);
          }
        }
        //s_type_group
        rs1 = pstmt_in_use_other.executeQuery();
        while (rs1.next()) {
          lid = rs1.getInt(1);
          tid = rs1.getInt(2);
          tag_name = rs1.getString(3);
          channel = rs1.getInt(4);

          ZTag2 zTag = new ZTag2(tid, tag_name, channel);
          processLogger
            .info("msgId:{},product id:{},tid:{},tag_name:{}", msgCount, lid, tid, tag_name);

          if (tagsMap.containsKey(lid)) {
            tagsMap.get(lid).add(zTag);
          } else {
            List<ZTag2> lt = new ArrayList<>();
            lt.add(zTag);
            tagsMap.put(lid, lt);
          }
        }
      } catch (Exception e) {
        flag = 0;
        errorLogger.info("test.msgId:{},DB failed,{}", msgCount, e.toString());
        e.printStackTrace();
      }finally {
        if(rs != null){
          try {
            rs.close();
          }catch (Exception e){
            errorLogger.info("msgId:{},DB resultSet failed,{}", msgCount, e);
          }
        }

        if(rs1 != null){
          try {
            rs1.close();
          }catch (Exception e){
            errorLogger.info("msgId:{},DB resultSet failed,{}", msgCount, e);
          }
        }
      }
      try {
        ////////////////////////////////////////
        long dbTime = System.currentTimeMillis();
        int dbCost = (int)(dbTime - now);
        ///////////////////////////////////////
        _outputCollector.emit(_streamId, tuple, new Values(index, type, tagsMap, msgCount, flag));
        ////////////////////////////////////////
        dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.OUT, msgCount, index, type, stage, tagsMap);
        ////////////////////////////////////////
        _outputCollector.ack(tuple);
        ////////////////////////////////////////
        long end = System.currentTimeMillis();
        int cost = (int)(end - now);
        ////////////////////////////////////////
        timeCostLogger.info("{}:{}:{}:{}:{}:{}", msgCount, now, dbTime, end, dbCost, cost);

      } catch (Exception e) {
        errorLogger.info("msgId:{},DB get tags failed,{}", msgCount, e);
        errorLogger.info("msgId:{},payload:{}", msgCount, payload);
        _outputCollector.fail(tuple);
      }
    }
  }

}
